package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

//将静态内容导入到当前类的空间，在当前类中直接使用静态方法或变量，无需指定 静态内容的类名
// 之前: Types.STRING   静态导入后: STRING
import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2023/3/31
 *

 *
 */
public class Demo8_LamdaTuple2Demo
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);
        SingleOutputStreamOperator<Tuple2<String, Integer>> ds2 = ds
            .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {

                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            })
            /*
                    returns(Class<T> typeClass) ： 仅仅适用于非泛型类型。
                               Collector<T>: T 对应 WordCount
                               Collector<T>: T 对应 Tuple2<String, Integer>
                               T 中还有泛型，无法使用 returns(Class<T> typeClass)。

                     推荐使用
                        returns(TypeHint<T> typeHint)
                            或
                        returns(TypeInformation<T> typeInfo)
                            工具类： Types，可以方便地返回 指定类型的 TypeInformation 对象。

             */
            //.returns(new TypeHint<Tuple2<String, Integer>>() {});
            .returns(TUPLE(STRING,INT));
        ds2
            .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
            .sum(1)
            .print();


        env.execute();

    }
}
